---
title: CompletableFuture
---

用多线程优化性能，其实不过就是将串行操作变成并行操作。在将串行操作变成并行操作的过程中，一定会涉及异步化。

例如下面的示例代码，为了提升性能，我们得把它们并行化。
```java
//以下两个方法都是耗时操作
doBizA();
doBizB();
```
还是很简单的，创建两个子线程去执行就可以了。

```java title='主线程无需等待 doBizA() 和 doBizB() 的结束结果，也就是说这两个操作已经被异步化了'
new Thread(()->doBizA()).start();
new Thread(()->doBizB()).start();
```

**异步化**，是并行方案得以实施的基础，更深入的讲就是：**利用多线程优化性能这个核心方案得以实施的基础**。

## CompletableFuture 的核心优势



使用 CompletableFuture 实现烧水泡茶程序。首先需要完成分工方案，分为 3 个任务：任务 1 负责洗水壶、烧开水，任务 2 负责洗茶壶、洗茶杯和拿茶叶，任务 3 负责泡茶。分工图如下所示：



![b33f823a4124c1220d8bd6d91b877e78](https://img.wkq.pub/pic/b33f823a4124c1220d8bd6d91b877e78.webp)

下面是代码实现：
```java
public class Test {
    public static void main(String[] args) {
        //任务一： 洗水壶 -> 烧开水
        CompletableFuture<Void> f1 = CompletableFuture.runAsync(() -> {
            try {
                System.out.println("T1:洗水壶...");
                Thread.sleep(1, TimeUnit.SECONDS.ordinal());
                System.out.println("T1:烧开水...");
                Thread.sleep(15, TimeUnit.SECONDS.ordinal());
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        //任务二： 洗茶壶 -> 洗茶杯 -> 取茶叶
        CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("T2:洗茶壶...");
                Thread.sleep(1, TimeUnit.SECONDS.ordinal());
                System.out.println("T2:洗茶杯...");
                Thread.sleep(2, TimeUnit.SECONDS.ordinal());
                System.out.println("T2:拿茶叶...");
                Thread.sleep(1, TimeUnit.SECONDS.ordinal());
                return "龙井";
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        //任务三： 泡茶
        CompletableFuture<String> f3 = f1.thenCombine(f2, (__, tf) -> {
            System.out.println("T1:拿到茶叶：" + tf);
            System.out.println("T1:泡茶...");
            return "上茶:" + tf;
        });
        //等待任务 3 执行结果
        System.out.println(f3.join());
    }
}
```
## 创建 CompletableFuture 对象

创建 CompletableFuture 对象主要靠下面代码中展示的这 4 个静态方法，在烧水泡茶的例子中，已经使用了 runAsync(Runnable runnable) 和 `supplyAsync(Supplier<U> supplier)`, 它们之间的区别是：Runnable 接口的 run() 方法没有返回值，而 Supplier 接口的 get() 方法是有返回值的。

前两个方法和后两个方法的区别在于：后两个方法可以指定线程池参数。

默认情况下 CompletableFuture 会使用公共的 ForkJoinPool 线程池，这个线程池默认创建的线程数是 CPU 的核数（也可以通过 JVM option: -Djava.util.concurrent.ForkJoinPool.common.parallelism 来设置 ForkJoinPool 线程池的线程数）。如果所有 CompletableFuture  共享一个线程池，那么一旦有任务执行一些很慢的的 I/O 操作，就会导致线程池中所有线程都阻塞在 I/O 操作上，从而造成线程饥饿，从而影响系统的性能。

所以，强烈建议要**根据不同的业务类型创建不同的线程池，以避免互相干扰。**

```java
//使用默认线程池
static CompilableFuture<Void> runAsync(Runnable runable);
static <U> CompteableFuture<U> supplyAsync(Suppliy<U> supplier);
//可以指定线程池
static CompletableFutre<Void> runAsync(Runnable runnable,Executor executor);
static CompletableFutre<Void> runAsync(Runnable runnable,Executor executor);
```

创建完 CompletableFuture 对象之后，会自动执行异步执行 runnable.run() 方法或者 supplier.get() 方法，对于一个异步操作，你需要关注两个问题：一个是异步操作什么时候结束，另一个是如何获取异步操作的结果。

因为 CompletableFuture 类实现了 Future 接口，所以这两个问题都可以通过 Future 接口来解决。另外，CompletableFuture 类还实现了 CompletionStage接口。

## 如何理解 CompletionStage接口
任务是有时序关系的，比如有**串行关系、并行关系、汇聚关系**等.

串行关系

![并行关系](https://img.wkq.pub/pic/e18181998b82718da811ce5807f0ad9f.webp)

并行关系

![ea8e1a41a02b0104b421c58b25343bd2](https://img.wkq.pub/pic/ea8e1a41a02b0104b421c58b25343bd2.webp)

汇聚关系

![3f1a5421333dd6d5c278ffd5299dc33b](https://img.wkq.pub/pic/3f1a5421333dd6d5c278ffd5299dc33b.webp)

 CompletionStage接口可以清晰地描述任务之间的这种时序关系，例如前面提到的 f3 = f1.thenCombine(f2,()->{})描述的就是一种汇聚关系。烧水泡茶程序中的汇聚关系是一种 AND 聚合关系，既然有 AND ，那就一定有 OR 聚合关系，所谓的 OR 聚合关系就是指依赖的任务只要有一个完成就可以执行当前任务。



CompletionStage接口也可以方便地描述异常处理。

### 1. 描述串行关系

CompletionStage接口里描述串行关系，主要是 thenApply、thenAccept、thenRun  和 thenCompose  这四个系列的接口。

thenApply 系列函数里参数 fn 的类型是接口 `Function<T,R>`,这个接口里与 CompletionStage 相关的方法是 R apply(T t)，这个方法既能接收参数也支持返回值，所以 thenApply 系列方法的返回值是 `CompletionStage<R>`

而 thenAccept 系列方法里参数 consumer 的类型是 接口` Consumer<T>`,这个接口里与 CompletionStage 相关的方法是 void accept(T t),这个方法虽然支持参数，但却不支持返回值，所以 thenAccept 系列的方法返回的是 `CompletionStage<Void>`。

thenRun 系列方法里的 action 参数的类型是 Runnable，所以 action 既不能接收参数也不支持返回值，所以 thenRun 系列方法返回的也是 `CompletionStage<Void>`.



这些方法里 Async 代表的是异步执行 fn、consumer 或者 action。其中，需要注意的是 thenCompose 系列方法，这个系列的方法会创建出一个子流程，最终结果和 thenApply 系列是相同的。

```java
CompletionStage<R> thenApply(fn);
CompletionStage<R> thenApplyAsync(fn);
CompletionStage<Void> thenAccept(cosumer);
CompletionStage<Void> thenAcceptAsync(consumer)
CompletionStage<Void> thenRun(action);
CompletionStage<R> thenCompose(fn);
CompletionStage<R> thenComposeAsync(fn);
```
首先通过 supplyAsync() 启动一个异步流程，之后是两个串行操作。不过，虽然这是一个异步操作，但任务①②③却是串行执行的，②依赖①的执行结果，③依赖②的执行结果
```java
public class Test {
    public static void main(String[] args) {
        CompletableFuture<String> fo = CompletableFuture.supplyAsync(() -> "Hello World")
                .thenApply(s -> s + "  QQ")
                .thenApply(String::toUpperCase);

        System.out.println(fo.join());
    }
}
```

### 2. 描述AND关系
CompletionStage 接口里面描述 AND 汇聚关系，主要是通过 thenCombine、thenAcceptBoth、runAfterBoth 系列的接口，这些接口的区别页源于 fn、consumer、action 这三个核心参数的不同。

```java
CompletionStage<R> thenCombine(other,fn);
CompletionStage<R> thenCombineAsync(other,fn);
CompletionStage<Void> thenAcceptBoth(other,consumer);
CompletionStage<Void> thenAcceptBothAsync(other,consumer);
CompletionStage<Void> runAfterBoth(other,runnable);
CompletionStage<Void> runAfterBoth(other,runnable);
```

### 3. 描述OR汇聚关系
CompletionStage 接口里面描述 OR 汇聚关系，主要是 ApplyToEither、acceptEither 和 runAfterEither 系列的接口，这些接口的区别也是源于 fn、consumer、action 这三个接口参数的不同。

```java
CompletionStage applyToEither(other,fn);
CompletionStage applyToEitherAsync(other,fn);
CompletionStage accetpEither(other,consumer);
CompletionStage accetpEitherAsync(other,runnable);
CompletionStage runAfterEither(other,consumer);
CompletionStage runAfterEitherAsync(other,runnable);
```
```java title='applyToEither()方法来描述一个 OR 汇聚关系'
public class Test {
    public static void main(String[] args) {
        System.out.println("getRandom(5,10) = " + getRandom(5, 10));
        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
            try {
                int t = getRandom(5, 10);
                sleep(t, TimeUnit.SECONDS.ordinal());
                return String.valueOf(t);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
            try {
                int t = getRandom(5, 10);
                sleep(t, TimeUnit.SECONDS.ordinal());
                return String.valueOf(t);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        CompletableFuture<String> f3 = f1.applyToEither(f2, s -> s);
        System.out.println(f3.join());
    }
}
```
### 4. 异常处理
fn 、consumer、action 它们的核心方法都**不允许抛出可检查异常，但是却无法限制它们抛出运行时异常**。在非异步编程里面，我们可以使用 try{}catch{} 来捕获并处理异常，那在异步编程里面该如何处理呢？
```java
CompletionFuture<Integer> f0 = CompletableFuture.supplyAsync(()->(7/0))
.thenApply(r->r*10);
System.out.println(f0.join());
```
CompletionStage 接口提供的方案非常简单，使用这些方法进行异常处理和串行操作是一样的，都支持链式编程方法。
```java
CompletionStage exceptionally(fn);
CompletionStage<R> whenComplete(consumer);
CompletionStage<R> whenCompleteAysnc(consumer);
CompletionStage<R> handle(fn);
CompletionStage<R> handleAsync(fn);
```
下面的示例代码展示了如何使用 exceptionally(fn) 方法来处理异常，exceptionally() 的使用非常类似于 try{}catch{} 中的 catch{}, 但是由于支持链式编程方式，所以相对简单。

whenComplete() 和 handle() 系列方法就类似于 try{}finally{} 中的 finally ，无论是否发生异常都会执行 whenComplete() 中的回调函数 consumer 和 handle() 中的回调函数 fn. whenComplete() 和 handle() 的区别在于 whenComplete() 不支持返回结果，而 handle() 是支持返回结果的。


```java
public class Test {
    public static void main(String[] args) {
        CompletableFuture<Integer> f0 = CompletableFuture.supplyAsync(() -> 7 / 0)
                .thenApply(r -> r * 10)
                .exceptionally(e -> 0);
        System.out.println(f0.join());
    }
}
```

不同线程之间的事务完全隔离，异步线程内仍是可以调用异步~



